/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: cbus_endpoint.c
 *
 * Purpose: endpoint implementation for cbus.
 *
 * Developer:
 *   wen.gu , 2022-01-18
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/
#include "cbus/bus/cbus_endpoint.h"

#include <string.h>

#include "cbus/core/cbus_mutex.h"
#include "cbus/bus/cbus_endpoint_ops.h"
#include "cbus/bus/cbus_message_packet.h"
#include "cbus/core/cbus_map.h"

#include "cbus/core/cbus_log.h"

/******************************************************************************
 **    MACROS
 ******************************************************************************/
#define CHECK_OPS_API(ep_ops, ops_api) if(!(ep_ops)->ops_api){ \
    LOGE("syntax error: endpoint ops must implementation API:'%s'\n", #ops_api ); \
    return GFALSE; \
}
/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/
typedef struct _endpoint_info
{
    cbus_message_packet_deserializer_t pkt_deserializer;
    cbus_endpoint_t ep_ref;    
}endpoint_info_t;

struct _cbus_endpoint
{
    const cbus_endpoint_ops* ops;
    GPHD handle;
    GPTR on_connect_opaque;
    //GPTR on_receive_opaque;
    cbus_endpoint_on_connect_changed_func on_connect;
    //cbus_message_on_process_func on_receive;
    cbus_message_manager_t  message_manager;
    cbus_map_t ep_info_map;
};
/******************************************************************************
 **    inner FUNCTION DEFINITIONS
 ******************************************************************************/

static GBOL cbus_endpoint_ops_check_valid(cbus_endpoint_ops* ops)
{
    CHECK_OPS_API(ops, initialize);
    CHECK_OPS_API(ops, finish);
    CHECK_OPS_API(ops, set_on_connect);
    CHECK_OPS_API(ops, set_on_receive);
    CHECK_OPS_API(ops, start);
    CHECK_OPS_API(ops, stop);
    CHECK_OPS_API(ops, send);

    return GTRUE;
}



static void endpoint_on_message(cbus_message_t request_message)
{
    cbus_endpoint_t ep = (cbus_endpoint_t)cbus_message_get_opaque(request_message);
#if 0
    LOGD("[%s.%d]receive message, msg id: %d, type: %d, ssid: %d\n", __FUNCTION__, __LINE__,
        cbus_message_get_message_id(request_message), cbus_message_get_message_type(request_message), cbus_message_get_session_id(request_message));    
#endif

    //LOGD("[%s.%d]===>: msg id: %d, type:%d, \n", __FUNCTION__, __LINE__, request_message->message_id, request_message->message_type);
    cbus_message_handler_t* msg_handler = cbus_message_manager_get_hander(ep->message_manager, cbus_message_get_message_type(request_message), cbus_message_get_message_id(request_message));

    if (msg_handler && msg_handler->on_message)
    {
        cbus_message_t message_ref = cbus_message_reference(request_message);
        cbus_message_set_opaque(message_ref, msg_handler->opaque);
        msg_handler->on_message(message_ref);
    }
    else
    {
        cbus_message_t response_message = cbus_message_alloc_response(request_message, 0); /** no payload ,just message header */

        if (!response_message)
        {
            GErrc ret = cbus_endpoint_send(ep, response_message);

            if (G_OK != ret)
            {
                LOGE("send error code to remote endpoint failed(0x%08x)\n", ret);
            }            

            cbus_message_free(response_message);
        }
        else
        { /** response type is unknown mean that ,this request type maybe is a response type , so in this case ,we cann't auto send 
          *   reponse message to avoid endless loop between two endpoint 
          */
            LOGW("[%s.%d]receive a request msg type: %d, and not handler to process it, but we cann't auto send response message, because the response type of current message is unkown\n",
                __FUNCTION__, __LINE__, cbus_message_get_message_type(request_message));
            /** todo something. */
        }

    }    
}

static void endpoint_info_free(void* ptr)
{
    endpoint_info_t* ep_info = (endpoint_info_t*)ptr;

    cbus_message_packet_deserializer_destroy(ep_info->pkt_deserializer);
    free(ep_info);
}

static endpoint_info_t* endpoint_info_create(cbus_endpoint_t ep, cbus_endpoint_id endpoint_id)
{
    endpoint_info_t* ep_info = (endpoint_info_t*)malloc(sizeof(endpoint_info_t));

    if (ep_info)
    {
        ep_info->pkt_deserializer = cbus_message_packet_deserializer_create(endpoint_id, endpoint_on_message, ep);

        if (!ep_info->pkt_deserializer)
        {
            LOGE("create packet deserializer failed\n");
            free(ep_info);
            return NULL;
        }

        ep_info->ep_ref = ep;
    }

    return ep_info;
}

static void cbus_endpoint_on_receive(GPTR opaque, cbus_endpoint_id endpoint_id, const GBUF data, GU32 size)
{
    //LOGD("[%s.%d]===>: ep id: %llu, size:%d\n", __FUNCTION__, __LINE__, (GU64)endpoint_id, size);
    endpoint_info_t* ep_info = (endpoint_info_t*)opaque;

    cbus_endpoint_t ep = ep_info->ep_ref;

    GErrc ret = cbus_message_packet_deserializer_deserialize(ep_info->pkt_deserializer, data, size);

    if (G_OK != ret)
    {
        LOGE("deserialize endpoint received data failed(0x%08x)\n", ret);
        //todo refine me , send response error code to remote endpoint        
    }
}


static void endpoint_info_map_add(cbus_endpoint_t ep, cbus_endpoint_id endpoint_id)
{
    endpoint_info_t* ep_info = NULL;
    if (cbus_map_find_pointer(ep->ep_info_map, (GU64)endpoint_id, (const void**)&ep_info) == GFALSE)
    {
        ep_info = endpoint_info_create(ep, endpoint_id);

        if (ep_info)
        {
            if (cbus_map_set_pointer(ep->ep_info_map, (GU64)endpoint_id, ep_info, endpoint_info_free) == GTRUE)
            {
                GErrc ret = ep->ops->set_on_receive(ep->handle, endpoint_id, cbus_endpoint_on_receive, ep_info);

                if (G_OK != ret)
                {
                    LOGE("set on receive handler failed(0x%08x)\n", ret);
                }
            }
            else
            {
                LOGE("insert ep info to map failed\n");
                endpoint_info_free(ep_info);
            }
        }
    }
    else
    {
        LOGW("warning, this endpoint id(%p) already exist\n", endpoint_id);
    }
}



static void cbus_endpoint_on_connect(GPTR opaque, cbus_endpoint_id endpoint_id, GBOL is_connected)
{
    cbus_endpoint_t ep = (cbus_endpoint_t)opaque;
    if (is_connected == GTRUE)
    {
        endpoint_info_map_add(ep, endpoint_id);        
    }
    else
    {
        cbus_map_remove(ep->ep_info_map, (GU64)endpoint_id);
    }

    if (ep->on_connect)
    {
        ep->on_connect(ep->on_connect_opaque, endpoint_id, is_connected);
    }
}


/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/

GErrc cbus_endpoint_create(cbus_endpoint_t* endpoint, cbus_endpoint_role role, const GSTR name, const GSTR url)
{
    if (!endpoint || !url)
    {
        return G_ErrBadParameter;
    }

    const cbus_endpoint_ops* ops = cbus_endpoint_ops_find(role, name);

    if (!ops)
    {
        LOGE("not found endpoint ops with(name: %s, role:%d)\n", name ? name : "null", role);
        return G_ErrNotFound;
    }

    if (cbus_endpoint_ops_check_valid(ops) == GFALSE)
    {
        return G_ErrUndefined;
    }

    cbus_endpoint_t ep = (cbus_endpoint_t)malloc(sizeof(struct _cbus_endpoint));

    if (!ep)
    {
        LOGE("alloc endpoint instance for (%s, role:%d) failed\n", url, role);
        return G_ErrInsufficientResources;
    }

    memset(ep, 0, sizeof(struct _cbus_endpoint));

    ep->ops = ops;
    ep->message_manager = cbus_message_manager_create();

    if (!ep->message_manager)
    {
        LOGE("alloc message manager instance for endpoint(%s, role:%d) failed\n", url, role);
        cbus_endpoint_destroy(ep);
        return G_ErrInsufficientResources;
    }

    ep->ep_info_map = cbus_map_create(CMV_TYPE_Pointer, CBUS_MAP_DEFAULT_CAPACITY);

    if (!ep->ep_info_map)
    {
        LOGE("create ep info map instance for endpoint(%s, role:%d) failed\n", url, role);
        cbus_endpoint_destroy(ep);
        return G_ErrInsufficientResources;
    }
    
    GErrc ret = ep->ops->initialize(&ep->handle, url);
    if (G_OK != ret)
    {
        LOGE("endpoint ops initialize failed(0x%08x)\n", ret);
        cbus_endpoint_destroy(ep);
        return ret;
    }


    ret = ep->ops->set_on_connect(ep->handle, cbus_endpoint_on_connect, ep);
    if (G_OK != ret)
    {
        LOGE("endpoint ops set_on_connect failed(0x%08x)\n", ret);
        cbus_endpoint_destroy(ep);
        return ret;
    }

    *endpoint  = ep;  
    return G_OK;
}

void cbus_endpoint_destroy(cbus_endpoint_t endpoint)
{
    if (endpoint)
    {
        cbus_message_manager_destroy(endpoint->message_manager);

        endpoint->ops->finish(endpoint->handle);

        cbus_map_destroy(endpoint->ep_info_map);

        free(endpoint);    
    }
}

GErrc cbus_endpoint_start(cbus_endpoint_t endpoint)
{
    if (!endpoint)
    {
        return G_ErrBadParameter;
    }

    return endpoint->ops->start(endpoint->handle);
}
GErrc cbus_endpoint_stop(cbus_endpoint_t endpoint)
{
    if (!endpoint)
    {
        return G_ErrBadParameter;
    }

    return endpoint->ops->stop(endpoint->handle);
}



GErrc cbus_endpoint_set_on_connect(cbus_endpoint_t endpoint, cbus_endpoint_on_connect_changed_func on_connect, GPTR opaque)
{
    if (!endpoint || !on_connect)
    {
        return G_ErrBadParameter;
    }

    endpoint->on_connect = on_connect;
    endpoint->on_connect_opaque = opaque;
    return G_OK;    
}

GErrc cbus_endpoint_register_message(cbus_endpoint_t endpoint, const cbus_message_info_t* msg_info)
{
    if (!endpoint)
    {
        return G_ErrBadParameter;
    }

    return cbus_message_manager_register(endpoint->message_manager,msg_info);
}

GErrc cbus_endpoint_register_message_category(cbus_endpoint_t endpoint, cbus_message_type_t message_type, cbus_message_on_process_func on_message, GPTR opaque)
{
    if (!endpoint || !on_message)
    {
        return G_ErrBadParameter;
    }

    return cbus_message_manager_register_category(endpoint->message_manager, message_type, on_message, opaque);
}

    /**independent_worker:  GTRUE: endpoint will be alloc a independent worker thread to proccess this message group, GFALSE: will not 
     * 
 */
GErrc cbus_endpoint_register_message_group(cbus_endpoint_t endpoint, const cbus_message_group_t* message_group)
{
    if (!endpoint || !message_group)
    {
        return G_ErrBadParameter;
    }   

    return cbus_message_manager_register_group(endpoint->message_manager, message_group);
}

GErrc cbus_endpoint_unregister_message(cbus_endpoint_t endpoint, cbus_message_type_t message_type, cbus_message_id message_id)
{
    if (!endpoint)
    {
        return G_ErrBadParameter;
    }   

    return cbus_message_manager_unregister(endpoint->message_manager, message_type, message_id);
}

GErrc cbus_endpoint_unregister_message_category(cbus_endpoint_t endpoint, cbus_message_type_t message_type)
{
    if (!endpoint)
    {
        return G_ErrBadParameter;
    }   

    return cbus_message_manager_unregister_category(endpoint->message_manager, message_type);
}

GErrc cbus_endpoint_unregister_message_all(cbus_endpoint_t endpoint)
{
    if (!endpoint)
    {
        return G_ErrBadParameter;
    }   

    return cbus_message_manager_unregister_all(endpoint->message_manager);
}

GErrc cbus_endpoint_send(cbus_endpoint_t endpoint, cbus_message_t message)
{
    if (!endpoint || !message)
    {
        return G_ErrBadParameter;
    }   

    return endpoint->ops->send(endpoint->handle, cbus_message_get_endpoint_id(message), cbus_message_get_data(message), cbus_message_get_size(message));
}








